package com.noah2021.util;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.PosixParser;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.srvutil.ServerUtil;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.command.SubCommandException;
import org.apache.rocketmq.tools.command.topic.UpdateTopicSubCommand;

import java.util.*;

/**
 * fixme 类中的方法不确定能否支持多namesrv的情况！
 * @author Zephyr
 * @date 2019/12/21.
 */
public class RocketMQUtil {

    static  String localhost="localhost";
    static  String port="9876";
    static  String brokerAddr(String brokerAddr){
        return  "-b "+brokerAddr;
    }

    /**
     *  return  "--topic "+topic; 不行啊
     * @param topic
     * @return
     */
    static  String topic(String topic){
//        return  "--topic "+topic;
        return  "-t "+topic;
    }
//    https://www.jianshu.com/p/b30c15031896
    public static void main(String[] args) throws SubCommandException, InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException {


        String namesrvAddr=   localhost+":"+"9876";

        ClusterInfo clusterInfo = getClusterInfo(namesrvAddr);
//        HashMap<String, BrokerData> brokerAddrTable = clusterInfo.getBrokerAddrTable();
//        HashMap<String, Set<String>> clusterAddrTable = clusterInfo.getClusterAddrTable();
//        System.out.println("clusterInfo");
//        System.out.println(clusterInfo);
        print(clusterInfo);
//        brokerAddrTable
//        {}
//        clusterAddrTable
//        {}
    }

    static  void print(ClusterInfo clusterInfo){
        HashMap<String, BrokerData> brokerAddrTable = clusterInfo.getBrokerAddrTable();
        HashMap<String, Set<String>> clusterAddrTable = clusterInfo.getClusterAddrTable();
        System.out.println("brokerAddrTable");
        System.out.println(brokerAddrTable);
        System.out.println("clusterAddrTable");
        System.out.println(clusterAddrTable);
    }

  static   void  create() throws SubCommandException {

//        -b,--brokerAddr <arg> create topic to which broker
//-t,--topic <arg> topic name
//        int port=9876;
//        tion: connect to <localhost:9874>
//        connect to <localhost:10909 为什么他会-2
        String[] subargs = new String[] {
//                "-b 10.1.4.231:10911",
                brokerAddr(localhost+":10911"),
//                brokerAddr(localhost+":"+port),
//                brokerAddr(localhost+":"+"9876"),
                topic("stock"),
//                "-t unit-test-from-java-111",
                //"-r 8",
                //"-w 8",
                //"-p 6",
                //"-o false",
                //"-u false",
                //"-s false"
        };
        boolean isTopicCreated = RocketMQUtil.createTopic(subargs);
        //boolean isTopicCreated = RocketMQUtil.createTopic("10.1.4.231:10911", "", "testttttt");
        if (isTopicCreated) {
            System.err.println("topic create success");
        }
    }
    /**
     * 创建topic 可以自定义所有topic支持的参数
     * @param subargs updateTopic命名支持的所有参数选项
     * @return topic创建成功，返回 true
     * @throws SubCommandException
     */
    public static boolean createTopic(String[] subargs) throws SubCommandException {
        /*String[] subargs = new String[] {
                "-b 10.1.4.231:10911",
                "-t unit-test-from-java-1",
                "-r 8",
                "-w 8",
                "-p 6",
                "-o false",
                "-u false",
                "-s false"};*/
        System.out.println("subargs");
        System.out.println(Arrays.toString(subargs));
        UpdateTopicSubCommand cmd = new UpdateTopicSubCommand();
        Options options = ServerUtil.buildCommandlineOptions(new Options());
        final Options updateTopicOptions = cmd.buildCommandlineOptions(options);
      String appName=  "mqadmin "  + cmd.commandName();
        System.out.println("appName");
        System.out.println(appName);
//        org.apache.commons.cli.UnrecognizedOptionException: Unrecognized option: --topic stock
        try{
//            SServerUtil.parseCmdLine()
            final CommandLine commandLine = SServerUtil
                    .parseCmdLine(appName,
                            subargs, updateTopicOptions, new PosixParser());
//            List argList = commandLine.getArgList();
//            final CommandLine commandLine = ServerUtil
//                    .parseCmdLine(appName,
//                            subargs, updateTopicOptions, new PosixParser());
            List argList = commandLine.getArgList();
            System.out.println("argList");
            System.out.println(argList);
            System.out.println("commandLine");
            System.out.println(commandLine);
//            Caused by: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to <localhost:10909> failed
//            commandLine.toString();
//        他怎么是个null
            cmd.execute(commandLine, updateTopicOptions, null);
        }catch (Exception e){
            e.printStackTrace();
        }


        return true;
    }

    /**
     * 根据 brokerAddr or clusterName 创建topic
     * @param brokerAddr  在指定 broker 上创建topic时，此参数为必填，否则传null
     * @param clusterName 在指定 cluster 上创建topic时，此参数为必填，否则传null
     * @param topic       要创建的topic
     * @return            创建成功，返回true
     */
    public static boolean createTopic(String brokerAddr, String clusterName, String topic) throws Exception {
        if (StringUtils.isBlank(topic)) {
            return false;
        }
        List<String> argList = new LinkedList<>();
        argList.add("-t "  + topic);
        if (StringUtils.isNotBlank(brokerAddr)) {
            argList.add("-b "+   brokerAddr.trim());
        } else {
            argList.add("-c "  + clusterName.trim());
        }
        return createTopic(argList.toArray(new String [0]));
    }

    /**
     * 在指定name server下使用默认参数创建topic
     * @param namesrvAddr
     * @param topic
     * @return
     */
    public static boolean createTopic(String namesrvAddr, String topic) {
        try {
            Set<String> clusterNames = RocketMQUtil.getClusterNames(namesrvAddr);
            for (String clusterName : clusterNames) {
                RocketMQUtil.createTopic(null, clusterName, topic);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 获取指定 namesrv下的集群信息
     * @param namesrvAddr
     * @return
     * @throws MQClientException
     * @throws InterruptedException
     * @throws MQBrokerException
     * @throws RemotingTimeoutException
     * @throws RemotingSendRequestException
     * @throws RemotingConnectException
     */
    public static ClusterInfo getClusterInfo(String namesrvAddr) throws MQClientException, InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
        if (StringUtils.isBlank(namesrvAddr)) {
            return new ClusterInfo();
        }
        DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt(5000L);
        mqAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
        mqAdminExt.setNamesrvAddr(namesrvAddr);
        mqAdminExt.start();

        ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
        mqAdminExt.shutdown();
        return clusterInfo;
    }

    /**
     * 获取指定name server下的所有集群名称
     * @param namesrvAddr
     * @return
     * @throws MQClientException
     * @throws InterruptedException
     * @throws MQBrokerException
     * @throws RemotingTimeoutException
     * @throws RemotingSendRequestException
     * @throws RemotingConnectException
     */
    public static Set<String> getClusterNames(String namesrvAddr) throws MQClientException, InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
        return getClusterInfo(namesrvAddr).getClusterAddrTable().keySet();
    }

    /**
     * 获取指定 namesrv 下的所有broker信息（多name server下不确定能否正常工作）
     * @param namesrvAddr namesrv地址
     * @return HashMap<String, BrokerData>
     */
    public static Map<String, BrokerData> getAllBrokerInfo(String namesrvAddr) throws MQClientException, InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
        return getClusterInfo(namesrvAddr).getBrokerAddrTable();
    }

    /**
     * 获取连接到指定 namesrv 下的所有broker地址
     * @param namesrvAddr
     * @return
     */
    public static Set<String> getBrokerAddrs(String namesrvAddr) throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException {
        Map<String, BrokerData> allBrokerInfo = getAllBrokerInfo(namesrvAddr);
        Set<String> brokerAddrs = new HashSet<>();
        for (BrokerData brokerData : allBrokerInfo.values()) {
            brokerAddrs.addAll(brokerData.getBrokerAddrs().values());
        }
        return brokerAddrs;
    }
}
